package com.shujia.akka.register

import akka.actor.{Actor, ActorRef, ActorSelection, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}

import java.util.Date
import java.util.concurrent.TimeUnit
import scala.concurrent.duration.FiniteDuration


object Worker extends App {
  var (masterHost, masterPort, workerHost, workerPort) = ("127.0.0.1", 8888, "127.0.0.1", 9999)

  private val config: Config = ConfigFactory.parseString(
    s"""
       |akka.actor.provider="akka.remote.RemoteActorRefProvider"
       |akka.remote.netty.tcp.hostname=$workerHost
       |akka.remote.netty.tcp.port=$workerPort
       |""".stripMargin
  )

  private val workerSystem: ActorSystem = ActorSystem("workerSystem", config)

  private val workerRef: ActorRef = workerSystem.actorOf(Props(new Worker(masterHost, masterPort)), "worker")

  workerRef ! "start"

}


class Worker(masterHost: String, masterPort: Int) extends Actor {

  private val masterRef: ActorSelection = context.actorSelection(s"akka.tcp://MasterSystem@$masterHost:$masterPort/user/master")
  private var id: String = "1001"

  override def receive: Receive = {
    case "start" => {
      println("worker初始化完成...")
      self ! "register"
    }
    case "register" => {
      println("worker开始注册...")
      val workerInfo: WorkerInfo = WorkerInfo(id, 8, 16, new Date(), new Date())
      masterRef ! workerInfo
      self ! workerInfo
    }

    case WorkerInfo(id, cpu, memory, registerDate, heartbeatDate) => {
      // 实现定时发送消息给Master
      context.system.scheduler.schedule(
        new FiniteDuration(0, TimeUnit.SECONDS)
        , new FiniteDuration(4, TimeUnit.SECONDS)
        , self
        , "heartbeat")(context.dispatcher)
      println("当前定时任务已经启动，每隔4秒发送一次心跳 ")

    }

    case "heartbeat" => {
      val heartbeat: WorkerHeartbeat = WorkerHeartbeat(id, new Date())
      masterRef ! heartbeat
    }
  }

}
